package com.fire.rabbitmq.ack;

import com.rabbitmq.client.ReturnCallback;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import javax.annotation.PostConstruct;

/**
 * @author ：Mr.Fire
 * @date ：Created in 2021/5/10 22:03
 * @Description 消息发送确认
 * <p>
 * ConfirmCallback  只确认消息是否正确到达 Exchange 中
 * ReturnCallback   消息没有正确到达队列时触发回调，如果正确到达队列不执行
 * <p>
 * 1. 如果消息没有到exchange,则confirm回调,ack=false
 * 2. 如果消息到达exchange,则confirm回调,ack=true
 * 3. exchange到queue成功,则不回调return
 * 4. exchange到queue失败,则回调return
 */
@Configuration
public class MessageAckHandler{


    private static final Logger log = LoggerFactory.getLogger(MessageAckHandler.class);

    @Bean("fireRabbitTemplate")
    public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory){
        RabbitTemplate  rabbitTemplate= new RabbitTemplate(connectionFactory);
        /**
         * 如果消息到达 exchange, 则 confirm 回调, ack = true
         * 如果消息不到达 exchange, 则 confirm 回调, ack = false
         */
        rabbitTemplate.setConfirmCallback((correlationData,ack,cause)->{
            log.info("收到回调:{}", ack == true ? "消息成功到达交换机" : "消息到达交换机失败");
            if (!ack) {
                log.info("correlationData:{}", correlationData.getId());
                log.info("消息到达交换机失败原因:{}", cause);
                // 根据业务逻辑实现消息补偿机制
            }else{
                log.info("消息发送到交换机成功！");
            }

        });
        /**
         * exchange 到达 queue, 则 returnedMessage 不回调
         * exchange 到达 queue 失败, 则 returnedMessage 回调
         * 需要设置spring.rabbitmq.publisher-returns=true
         */
        rabbitTemplate.setReturnsCallback(returnedMessage->{
            log.info("消息未到达队列，setReturnsCallback回调");
            log.info("消息报文:{}", new String(returnedMessage.getMessage().getBody()));
            log.info("消息编号:{}", returnedMessage.getReplyCode());
            log.info("描述:{}", returnedMessage.getReplyText());
            log.info("交换机名称:{}", returnedMessage.getExchange());
            log.info("路由名称:{}", returnedMessage.getRoutingKey());
            // 根据业务逻辑实现消息补偿机制
        });
        return rabbitTemplate;
    }


}
